using Org.Apache.Rocketmq;
using Vin.Extension.RocketMQ.Options;
using Vin.Tool.AspNetCore.Singleton;
using Vin.Tool.Core.AssertCategory;

namespace Vin.Extension.RocketMQ.Models;

public abstract class VinConsumer : IMessageListener
{
    protected abstract string Group { get; }

    protected abstract Dictionary<string, FilterExpression> FilterExpressions { get; }

    protected virtual ClientConfig? ClientConfig { get; private set; }

    protected virtual int? ConsumptionThreadCount { get; } = null;

    protected virtual int? MaxCacheMessageCount { get; } = null;

    protected virtual int? MaxCacheMessageSizeInBytes { get; } = null;

    protected PushConsumer? Consumer;

    internal async Task Init()
    {
        if (Group.IsNullOrWhiteSpace()) throw new ArgumentNullException(nameof(Group));
        this.ClientConfig ??= VinApp.GetConfig<RocketMqOption>().ClientConfig;
        var builder = new PushConsumer.Builder()
            .SetClientConfig(this.ClientConfig)
            .SetConsumerGroup(Group)
            .SetMessageListener(this);
        if (FilterExpressions.Count > 0) builder.SetSubscriptionExpression(FilterExpressions);
        if (ConsumptionThreadCount.HasValue) builder.SetConsumptionThreadCount(ConsumptionThreadCount.Value);
        if (MaxCacheMessageCount.HasValue) builder.SetMaxCacheMessageCount(MaxCacheMessageCount.Value);
        if (MaxCacheMessageSizeInBytes.HasValue)
            builder.SetMaxCacheMessageSizeInBytes(MaxCacheMessageSizeInBytes.Value);
        Consumer = await builder.Build();
    }

    public void Dispose()
    {
        Consumer?.Dispose();
    }

    public abstract ConsumeResult Consume(MessageView messageView);
}